package com.intellif.mozping.util;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.Properties;

public class LogProducer<K,V> {
	private final Producer<K, V> producer;

	public LogProducer(Properties config) {
		this.producer = new KafkaProducer<K, V>(config);
		Runtime.getRuntime().addShutdownHook(new Thread() {
			@Override
			public void run() {
				producer.close();
			}
		});
	}

	public void send(String topic, K k, V v) {
		producer.send(new ProducerRecord<K, V>(topic, k, v));
	}

	public void send(String topic, V v) {
		producer.send(new ProducerRecord<K, V>(topic, v));
	}
	
	public void close(){
		producer.flush();
		producer.close();
	}

	/*public static void main(String[] args) {
		Properties properties = TaskUtil.getKafkaConf(Constant.properties_url);
		LogProducer<String, String> producer = new LogProducer<>(properties);
		String topic = properties.getProperty("topicName");
		for (int i = 0; i < 100000; i++) {
			producer.send(topic, "value-" + i);
		}
	}*/
}
